package org.apache.flink.state

import org.apache.flink.api.common.functions.{AggregateFunction, ReduceFunction, RichFlatMapFunction}
import org.apache.flink.api.common.state.{AggregatingState, AggregatingStateDescriptor, ReducingState, ReducingStateDescriptor}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

/**
 * 需求：求和
 * (1, Contains: 3 and 5 and 7)
 * (2, Contains: 4 and 2 and 5)
 *
 **/
object KeyedStateApp3 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.fromCollection(List(
      (1L, 3L),
      (1L, 7L),
      (2L, 4L),
      (1L, 5L),
      (2L, 2L),
      (2L, 5L)
    )).keyBy(_._1)
        .flatMap(new RichFlatMapFunction[(Long,Long),(Long, String)] {
          private var aggregatingState: AggregatingState[Long,String] = _

          override def open(parameters: Configuration): Unit = {

            val descriptor = new AggregatingStateDescriptor[Long, String, String]("contains",
              new AggregateFunction[Long, String, String] {
                override def createAccumulator(): String = "Contains:"

                override def add(value: Long, accumulator: String): String = {
                  if("Contains:" == accumulator) {
                    accumulator + value
                  } else {
                    accumulator + " and " + value
                  }
                }

                override def getResult(accumulator: String): String = accumulator

                override def merge(a: String, b: String): String = a + " and " + b
              }
            ,classOf[String]
            )

            aggregatingState = getRuntimeContext.getAggregatingState(descriptor)
          }

          override def flatMap(value: (Long, Long), out: Collector[(Long, String)]): Unit = {
            aggregatingState.add(value._2)
            out.collect(value._1, aggregatingState.get())
          }
        })
        .print()

    env.execute(getClass.getCanonicalName)
  }
}

